/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.client;

import org.slf4j.Logger;
import org.sss.mapreduce.GroupID;
import org.sss.mapreduce.datatype.Packable;

/**
 * <code>TupleGroup</code> has data set status in a Job.
 */
class TupleGroupStatus {
  private static Logger logger = SssClient.getLogger(TupleGroupStatus.class);
  private final GroupID id;
  private int upperCount = 0;
  private int doneCount = 0;
  private Class<? extends Packable> keyClass = null;
  private Class<? extends Packable> valueClass = null;

  /**
   * Allocates a <code>TupleGroup</code> object and initializes it so that
   * it has the given name and internal <code>GroupID</doce> object.
   *
   * @param  gid   the internal <code>GroupID</code> for this <code>TupleGroup</code>
   * @see    org.sss.mapreduce.GroupID
   *
   * NOTE: This used from only this module.
   */
  TupleGroupStatus(GroupID gid) {
    this.id = gid;
  }
  /**
   * Indicates whether this <code>TupleGroup</code> is ready to be read.
   *
   * @return true if all data output to this <code>TupleGroup</code> are completed.  otherwise false.
   *
   * NOTE: This used from only this module.
   */
  boolean hasData() {
    return doneCount == upperCount;
  }

  /**
   * Increment <code>upperCount</code> private variable by one.
   *
   * This means that a job is connected this <code>TupleGroup</code> in the job tree,
   * which write tuples toward this <code>TupleGroup</code>.
   *
   * NOTE: This used from only this module.
   */
  synchronized void incUpper() {
    upperCount++;
  }
  /**
   * Increment <code>doneCount</code> private variable by one.
   * This means that a job has finished writing tuples to this <code>TupleGroup</code>.
   *
   * NOTE: This used from only this module.
   */
  synchronized void incDone() {
    doneCount++;
  }

  /**
   * Reset <code>doneCount</code> private varible to 0.
   *
   * NOTE: This used from only this module.
   */
  synchronized void reset() {
    doneCount = 0;
  }

  @Override
  public String toString() {
    return id.toString() + "(" + doneCount + "/" + upperCount + ")";
  }
  /**
   * Indicates whether this <code>TupleGroup</code> is typed or not.
   * Typed <code>TupleGroup</code> object has been assigned valid key class
   * object and value class object pair.
   *
   * @return true if this <code>TupleGroup</code> has correct key class and value class pair. otherwise false.
   *
   * NOTE: This used from only this module.
   */
  boolean isTyped() {
    assert((keyClass == null && valueClass == null) || (keyClass != null && valueClass != null));
    return keyClass != null;
  }
  /**
   * Return the key class object for this <code>TupleGroup</code>.
   *
   * @return the key class object for this <code>TupleGroup</code>
   */
  Class<? extends Packable> getKeyClass() {
    return keyClass;
  }
  /**
   * Return the value class object for this <code>TupleGroup</code>.
   *
   * @return the value class object for this <code>TupleGroup</code>
   */
  Class<? extends Packable> getValueClass() {
    return valueClass;
  }
  /**
   * Set the key class and value class of this <code>TupleGroup</code>.
   *
   * @param  keyClass    the class object to be set as key class of this <code>TupleGroup</code>
   * @param  valueClass  the class object to be set as value class of this <code>TupleGroup</code>
   *
   * @return true - key class and value class are set as expected, otherwise false
   *
   * NOTE: This used from only this module.
   */
  boolean setOrCheckTypes(Class<? extends Packable> keyClass,
                          Class<? extends Packable> valueClass) {
    assert(keyClass != null);
    assert(valueClass != null);

    if (this.keyClass != null) {
      assert(valueClass != null);
      return this.keyClass == keyClass && this.valueClass == valueClass;
    }
    else {
      assert(this.valueClass == null);
      logger.debug("set types of " + id + ": {}, {}.",
                   keyClass.getName(), valueClass.getName());
      this.keyClass   = keyClass;
      this.valueClass = valueClass;
      return true;
    }
  }
  /**
   * Returns the internal <code>GroupID</code> object of this <code>TupleGroup</code>.
   *
   * @return the <code>GroupID</code> object
   * @see    org.sss.mapreduce.GroupID
   */
  public GroupID getGroupID() {
    return id;
  }
}
